package com.hmall.common.helper;

import com.hmall.common.utils.UserContext;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

@Slf4j
@RequiredArgsConstructor
@Component
public class RabbitMqHelper {

    private final RabbitTemplate rabbitTemplate;

    /**
     * 发送普通消息
     *
     * @param exchange
     * @param routingKey
     * @param msg
     */
    public void sendMessage(String exchange, String routingKey, Object msg) {
        log.info("发送消息到MQ，msg：{}", msg);
        rabbitTemplate.convertAndSend(exchange, routingKey, msg, addUserInfoToHeader());
    }

    /**
     * 发送消息处理器处理过的消息
     *
     * @param exchange
     * @param routingKey
     * @param msg
     * @param messagePostProcessor
     */
    public void sendMessage(String exchange, String routingKey, Object msg, MessagePostProcessor messagePostProcessor) {
        rabbitTemplate.convertAndSend(exchange, routingKey, msg, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //先将用户信息添加到消息头
                message.getMessageProperties().setHeader("user-info", UserContext.getUser());
                //再调用传入的messagePostProcessor，返回处理后的message
                if (messagePostProcessor != null) {
                    return messagePostProcessor.postProcessMessage(message);
                }
                //若messagePostProcessor为空直接返回
                return message;
            }
        });
    }

    /**
     * 发送延迟消息
     *
     * @param exchange
     * @param routingKey
     * @param msg
     * @param delay
     */
    public void sendDelayMessage(String exchange, String routingKey, Object msg, int delay) {
        rabbitTemplate.convertAndSend(exchange, routingKey, msg
                , new MessagePostProcessor() {
                    @Override
                    public Message postProcessMessage(Message message) throws AmqpException {
                        //先将用户信息添加到消息头
                        message.getMessageProperties().setHeader("user-info", UserContext.getUser());
                        //设置延迟时间
                        message.getMessageProperties().setDelay(delay);
                        return message;
                    }
                });
    }

    /**
     * 发送确认消息
     *
     * @param exchange
     * @param routingKey
     * @param msg
     * @param maxRetries
     */
    public void sendMessageWithConfirm(String exchange, String routingKey, Object msg, int maxRetries) {
        // 1.创建CorrelationData
        CorrelationData cd = new CorrelationData();
        // 2.给Future添加ConfirmCallback
        cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
            @Override
            public void onFailure(Throwable ex) {
                log.info("spring amqp 处理确认结果异常", ex);
            }
            @Override
            public void onSuccess(CorrelationData.Confirm result) {
                //回调逻辑
                if (result.isAck()) {
                    log.info("发送消息成功，收到ack！");
                } else {
                    log.info("发送消息失败，收到 nack，reason：{}", result.getReason());
                }
            }
        });
        rabbitTemplate.convertAndSend(exchange, routingKey, msg, addUserInfoToHeader(), cd);
    }

    private MessagePostProcessor addUserInfoToHeader() {
        return new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setHeader("user-info", UserContext.getUser());
                return message;
            }
        };
    }
}